package net.bwie.realtime.vehicle.dwd.function;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.utils.KafkaUtil;
import net.bwie.realtime.vehicle.dwd.bean.AreaControl;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class VehicleDistribution {

    public static void main(String[] args) throws Exception{

        //1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取数据源
//        DataStream<String> carStream = KafkaUtil.consumerKafka(env, "dwd-traffic-monitor-log");
        DataStream<String> carStream = KafkaUtil.consumerKafka(env, "car_log");

        //3.数据处理
        DataStream<String> resultStream = handle(carStream);
        resultStream.print("result");
        //4.数据输出

        //5.执行
        env.execute("VehicleDistribution");
    }

    private static DataStream<String> handle(DataStream<String> stream) {
        //按照areaId进行分组，用于计算去重，用state状态来表示第一次访问
        KeyedStream<String, String> areaIdStream = stream.keyBy(
                json -> JSON.parseObject(json).getString("areaId")+","+JSON.parseObject(json).getString("car")
        );
//        areaIdStream.print("aa");
        // 将流中每条数据进行封装bean对象
        SingleOutputStreamOperator<AreaControl> beanStream = areaIdStream.map(new TrafficMapFunction());
        beanStream.print("bb");

        //设置分组
        KeyedStream<AreaControl, String> keyedStream = beanStream.keyBy(bean -> bean.getAreaId());
        //窗口 1s

        WindowedStream<AreaControl, String, TimeWindow> windowStream = keyedStream.window(
                TumblingEventTimeWindows.of(Time.seconds(1))
        );

        //聚合计算
        DataStream<String> reportStream = windowStream.reduce(
                new PageViewReportReduceFunction(),new PageViewReportWindowFunction()
        );

        return reportStream;
    }
}
